package com.diaohw.platform.utility.netty;

import cn.hutool.core.net.NetUtil;
import com.diaohw.platform.common.util.ThreadUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;

@Slf4j
public class NettyClient {
    private ChannelFuture channelFuture;
    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
    private volatile boolean closed = false;

    private final int port;

    private final String host;

    public NettyClient(int port, String host) {
        this.port = port;
        this.host = host;
    }

    public synchronized void connect(ChannelHandler handler) {


        CompletableFuture.runAsync(() -> {
            ping();
            // 创建EventLoopGroup，用于处理客户端的I/O操作
            EventLoopGroup groupThread = new NioEventLoopGroup();

            // 创建Bootstrap实例，客户端启动对象
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(groupThread);
            // 设置服务端Channel类型为NioSocketChannel作为通道实现
            bootstrap.channel(NioSocketChannel.class);
            // 设置客户端处理
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) {
                    socketChannel.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                    // LengthFieldPrepender是一个编码器，主要是在响应字节数据前面添加字节长度字段
                    socketChannel.pipeline().addLast(new LengthFieldPrepender(4));
                    //handler一般都是放在最后面,建议一个就可以,如果业务复杂拆分多个,那在handler中调用fireXXX才会向下一个handler继续调用
                    //StringDecoder要放到LengthFieldBasedFrameDecoder后面,对得到消息体的ByteBuf转码为String类型,个人认为这个比较常用
                    socketChannel.pipeline().addLast("encode", new StringDecoder());
                    socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                            reconnect(handler);
                        }
                    });
                    socketChannel.pipeline().addLast(handler);

                }
            });


            try {
                // 绑定端口
                channelFuture = bootstrap.connect(host, port).sync();

                channelFuture.addListener((ChannelFutureListener) future -> {
                    closed = !future.isSuccess();
                    if (future.isSuccess()) {
                        log.info("netty client started on {}:{}", host, port);
                        while (!queue.isEmpty()) {
                            sendMessage(queue.poll());
                        }

                    } else {
                        log.error("netty client start failed", future.cause());
                    }
                });


                channelFuture.channel().closeFuture().sync();

            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                // 优雅地关闭线程
                groupThread.shutdownGracefully();
            }
        });

    }

    private void ping() {
        InetSocketAddress address = new InetSocketAddress(host, port);
        while (!NetUtil.isOpen(address, 1000)) {
            log.error("netty Server IP:{}, Port:{} is closed,Reconnect now ...", host, port);
            ThreadUtil.sleep(1000);
        }
    }

    private void reconnect(ChannelHandler handler) {
        closed = true;
        ping();
        connect(handler);

    }


    public void sendMessage(String message) {
        if (closed) {
            queue.add(message);
            return;
        }

        channelFuture.channel().writeAndFlush(Unpooled.copiedBuffer(message, CharsetUtil.UTF_8));
    }


}
